package com.our.base.easylog.es;

import cn.hutool.core.lang.UUID;
import com.alibaba.fastjson.JSON;
import com.our.base.easylog.IPushLog;
import com.our.base.easylog.buffer.cache.CachePool;
import com.our.base.easylog.config.es.EsProperties;
import com.our.base.easylog.entity.LogInfo;
import com.our.base.easylog.es.document.LogDocument;
import lombok.extern.slf4j.Slf4j;
import org.elasticsearch.action.ActionListener;
import org.elasticsearch.action.index.IndexRequest;
import org.elasticsearch.action.index.IndexResponse;
import org.elasticsearch.action.support.WriteRequest;
import org.elasticsearch.action.support.replication.ReplicationResponse;
import org.elasticsearch.client.RequestOptions;
import org.elasticsearch.client.RestHighLevelClient;
import org.elasticsearch.common.xcontent.XContentType;
import org.elasticsearch.core.TimeValue;
import org.springframework.beans.factory.annotation.Autowired;

import java.util.List;

/**
 * es日志推送实现
 *
 * @author Alice
 * @date 2023/11/29
 **/
@Slf4j
public class EsPushLogImpl implements IPushLog {
    private final EsProperties esProperties;
    private CachePool cachePool = CachePool.getCachePool();

    @Autowired
    private RestHighLevelClient restHighLevelClient;

    public EsPushLogImpl(EsProperties esProperties) {
        this.esProperties = esProperties;
    }

    @Override
    public void pushLog() {
        //拿缓存池里面的数据
        List<Object> logInfoList = cachePool.getAll();
        this.cachePool.cleanCache();
        log.info("推送日志{}条" + logInfoList.size());
        if (logInfoList.size() <= 0) {
            return;
        }
        logInfoList.forEach(obj ->
                asyncWrite(new LogDocument(((LogInfo) obj).getLog(), ((LogInfo) obj).getLogTime()))
        );
    }

    public void asyncWrite(LogDocument document) {
        IndexRequest indexRequest = new IndexRequest("test_demo");
        indexRequest.source(JSON.toJSONString(document), XContentType.JSON);
        indexRequest.timeout(TimeValue.timeValueSeconds(1));
        indexRequest.setRefreshPolicy(WriteRequest.RefreshPolicy.WAIT_UNTIL);
        //数据为存储而不是更新
        indexRequest.create(true);
        indexRequest.id(UUID.fastUUID().toString());
        restHighLevelClient.indexAsync(indexRequest, RequestOptions.DEFAULT, new ActionListener<IndexResponse>() {
            @Override
            public void onResponse(IndexResponse indexResponse) {
                ReplicationResponse.ShardInfo shardInfo = indexResponse.getShardInfo();
                if (shardInfo.getFailed() > 0) {
                    for (ReplicationResponse.ShardInfo.Failure failure : shardInfo.getFailures()) {
                        log.error("将id为：{}的数据存入ES时存在失败的分片，原因为：{}", indexRequest.id(), failure.getCause());
                    }
                }
            }

            @Override
            public void onFailure(Exception e) {
                log.error("{}:存储es时异常，数据信息为", indexRequest.id(), e);
            }
        });
    }
}
